import akshare as ak
from datetime import date, datetime, timedelta
from typing import List, Dict, Optional
import logging
import asyncio

import pandas as pd

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class TradingDayChecker:
    """
    交易日检查工具类
    """
    _trade_dates = None
    
    @classmethod
    def _load_trade_dates(cls):
        """
        加载交易日历数据
        """
        if cls._trade_dates is None:
            try:
                # 获取历史交易日历数据
                df = ak.tool_trade_date_hist_sina()
                # 将日期转换为字符串列表，格式为 YYYYMMDD
                df["trade_date"] = pd.to_datetime(df["trade_date"])
                cls._trade_dates = set(df['trade_date'].dt.strftime('%Y%m%d'))
            except Exception as e:
                logger.error(f"加载交易日历数据失败: {e}")
                cls._trade_dates = set()
    
    @classmethod
    def is_trading_day(cls, check_date: str) -> bool:
        """
        判断指定日期是否为交易日
        
        Args:
            check_date: 日期，格式为 YYYYMMDD
            
        Returns:
            bool: True表示是交易日，False表示不是交易日
        """
        # 加载交易日历数据
        cls._load_trade_dates()
        
        # 直接在交易日历中查找
        return check_date in cls._trade_dates # type: ignore


async def fetch_stock_list() -> List[Dict]:
    """
    获取全量A股股票列表
    
    Returns:
        List[Dict]: 股票列表，包含股票代码和名称
    """
    try:
        logger.info("正在获取全量A股股票列表")
        
        # 使用 akshare 获取A股股票列表
        loop = asyncio.get_event_loop()
        stock_list_data = await loop.run_in_executor(None, ak.stock_info_a_code_name)
        
        # 将 DataFrame 转换为字典列表
        if stock_list_data is not None and not stock_list_data.empty:
            # 转换为记录列表
            result = stock_list_data.to_dict(orient="records")
            logger.info(f"成功获取 {len(result)} 只A股股票")
            return result
        else:
            logger.warning("未获取到A股股票列表")
            return []
            
    except AttributeError as e:
        logger.error(f"akshare 属性错误，请检查API是否变更: {str(e)}")
        return []
    except Exception as e:
        logger.error(f"获取A股股票列表时发生错误: {str(e)}")
        return []


async def fetch_zt_stock_data(query_date: Optional[str] = None) -> List[Dict]:
    """
    获取涨停股票数据
    
    Args:
        query_date: 查询日期，格式为 YYYYMMDD，如不提供则获取最新数据
        
    Returns:
        List[Dict]: 涨停股票数据列表
    """
    try:
        if query_date is None:
            # 如果没有提供日期，则使用今天日期
            query_date = datetime.now().strftime("%Y%m%d")
        
        logger.info(f"正在获取 {query_date} 的涨停股票数据")

        # 检查是否为交易日
        if not TradingDayChecker.is_trading_day(query_date):
            logger.warning(f"{query_date} 不是交易日，无法获取数据")
            return []
        
        # 使用 akshare 获取涨停股票数据
        # 注意：akshare 目前主要是同步库，如果需要异步执行，应该在执行池中运行
        loop = asyncio.get_event_loop()
        zt_data = await loop.run_in_executor(None, lambda: ak.stock_zt_pool_em(date=query_date))
        
        # 将 DataFrame 转换为字典列表
        if zt_data is not None and not zt_data.empty:
            # 转换为记录列表
            result = zt_data.to_dict(orient="records")
            logger.info(f"成功获取 {len(result)} 条涨停股票数据")
            return result
        else:
            logger.warning(f"未获取到 {query_date} 的涨停股票数据")
            return []
            
    except AttributeError as e:
        logger.error(f"akshare 属性错误，请检查API是否变更: {str(e)}")
        return []
    except Exception as e:
        logger.error(f"获取涨停股票数据时发生错误: {str(e)}")
        return []


async def fetch_daily_kline_data(code: str, start_date: str, end_date: str) -> List[Dict]:
    """
    获取股票日K线数据
    
    Args:
        code: 股票代码
        start_date: 开始日期，格式为 YYYYMMDD
        end_date: 结束日期，格式为 YYYYMMDD
        
    Returns:
        List[Dict]: 日K线数据列表
    """
    try:
        logger.info(f"正在获取股票 {code} 从 {start_date} 到 {end_date} 的日K线数据")
        
        # 使用 akshare 获取日K线数据
        loop = asyncio.get_event_loop()
        # 东财接口需要的日期格式为 YYYYMMDD
        kline_data = await loop.run_in_executor(
            None, 
            lambda: ak.stock_zh_a_hist(
                symbol=code, 
                period="daily", 
                start_date=start_date, 
                end_date=end_date,
                adjust=""
            )
        )
        
        # 将 DataFrame 转换为字典列表
        if kline_data is not None and not kline_data.empty:
            # 转换为记录列表
            result = kline_data.to_dict(orient="records")
            logger.info(f"成功获取 {len(result)} 条日K线数据")
            return result
        else:
            logger.warning(f"未获取到股票 {code} 从 {start_date} 到 {end_date} 的日K线数据")
            return []
            
    except AttributeError as e:
        logger.error(f"akshare 属性错误，请检查API是否变更: {str(e)}")
        return []
    except Exception as e:
        logger.error(f"获取股票 {code} 日K线数据时发生错误: {str(e)}")
        return []


async def fetch_minute_kline_data(code: str, period: str = "1", start_date: str = None, end_date: str = None) -> List[Dict]:
    """
    获取股票分钟K线数据
    
    Args:
        code: 股票代码
        period: 时间周期，可选 "1", "5", "15", "30", "60" 分钟
        start_date: 开始日期，格式为 YYYYMMDD
        end_date: 结束日期，格式为 YYYYMMDD
        
    Returns:
        List[Dict]: 分钟K线数据列表
    """
    try:
        logger.info(f"正在获取股票 {code} 的{period}分钟K线数据")
        
        # 使用 akshare 获取分钟K线数据
        loop = asyncio.get_event_loop()
        
        # 根据是否提供日期范围选择不同的接口
        if start_date and end_date:
            # 如果提供了日期范围，使用历史数据接口
            kline_data = await loop.run_in_executor(
                None,
                lambda: ak.stock_zh_a_hist_min_em(
                    symbol=code,
                    period=period,
                    start_date=f"{start_date}000000",
                    end_date=f"{end_date}235959",
                    adjust=""
                )
            )
        else:
            # 如果没有提供日期范围，使用近期数据接口
            kline_data = await loop.run_in_executor(
                None,
                lambda: ak.stock_zh_a_minute(
                    symbol=code,
                    period=period
                )
            )
        
        # 将 DataFrame 转换为字典列表
        if kline_data is not None and not kline_data.empty:
            # 转换为记录列表
            result = kline_data.to_dict(orient="records")
            logger.info(f"成功获取 {len(result)} 条{period}分钟K线数据")
            return result
        else:
            logger.warning(f"未获取到股票 {code} 的{period}分钟K线数据")
            return []
            
    except AttributeError as e:
        logger.error(f"akshare 属性错误，请检查API是否变更: {str(e)}")
        return []
    except Exception as e:
        logger.error(f"获取股票 {code} {period}分钟K线数据时发生错误: {str(e)}")
        return []


if __name__ == "__main__":
    # 异步函数需要在事件循环中运行
    async def main():
        result = await fetch_zt_stock_data("20251011")
        print(result)
    
    # 运行异步主函数
    # asyncio.run(main())

    # 循环测试最近15天
    async def main_15():
        for i in range(15):
            test_date = (date.today() - timedelta(days=i)).strftime("%Y%m%d")
            result = await fetch_zt_stock_data(test_date)
    # asyncio.run(main_15())
    df = ak.tool_trade_date_hist_sina()
    print(len(df))